package tk.xboot.flink.vs;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;


/**
 * 两个方法:open(),map()
 * <p>
 * 在open中，我们通过ValueStateDescriptor创建了我们需要的state
 * 这个方法的参数包含名字，而且提供了序列化需要的信息（MovingAverage.class）
 *
 * 这个map方法会用来让每个事件值更加平滑。每个事件都会调用一次map，
 * 当然这个事件是与一个特定的key相关（一个传感器），
 * valueState保存了某个传感器（key）的之前事件的信息，
 * 我们可以将本次事件的值与之前事件的传感器值联系起来，
 * 去一个更加平滑的值。可以通过取平均值。
 *
 * more details visit :
 * https://www.jianshu.com/p/18cf04e0d350
 *
 *
 */

public class AvgMap extends RichMapFunction<Tuple2<String, Double>, Tuple2<String, Double>> {
    private ValueState<Avg> averageState;

    @Override
    public void open(Configuration conf) {
        ValueStateDescriptor<Avg> descriptor = new ValueStateDescriptor<>("average", Avg.class);
        averageState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public Tuple2<String, Double> map(Tuple2<String, Double> item) throws Exception {
        // access the state for this key
        Avg average = averageState.value();

        // create a new MovingAverage (with window size 2) if none exists for this key
        if (average == null) average = new Avg(2);

        // add this event to the moving average
        average.add(item.f1);
        averageState.update(average);

        // return the smoothed result
        return new Tuple2(item.f0, average.getAverage());
    }

    public static class Avg {
        private double avg;
        private int count;

        public Avg(int val) {
            this.avg = val;
        }

        public void add(double newVal) {
            avg = avg + newVal;
            avg = avg / (++count);
        }

        public double getAverage() {
            return avg;
        }

    }
}